/**
 * Copyright (C) 2009+ Spark Zheng
 *
 * @file        src/filters/test_1/test_1.c 
 * @date        2010-05-04
 * @brief	filter_module
 * @see         dserver
 * @history
 */

#include <dlib/str.h>
#include <dlib/mempool.h>
#include <dlib/confloose.h>
#include <dlib/buffer.h>

#include "dserver.h"
#include "dserver_errno.h"
#include "dserver_log.h"
#include "test_1.h"

int filter_on_load(const void *settings, filter_node_t *filter)
{
	char *str;
	dlib_confl_t *confl;

	DLOG_DEBUG(LOG_FILT, "FILTER(test_1): filter_on_load.");

	if (settings == NULL || filter == NULL) {
		DLOG_ERR(LOG_FILT, "filter_on_load parameters error.");
		return -E_ARGS_ILLEGAL;
	}
	confl = ((const dserver_setting_t *)settings)->dserver_confl;

	str = dlib_confl_get_string(confl, filter->f_name, "data", CONFL_STR_MOVE);
	if (str != NULL) {
		filter->f_data = str;
	} else {
		filter->f_data = strdup("test_1:test data in default");
	}

	return PROTOCOL_SUCCESS;
}

/**
 * do some persistence operation,
 * like build up a persist-connection
 */
void *filter_stream_init(void *arg)
{
	DLOG_DEBUG(LOG_FILT, "FILTER(test_1): filter_stream_init.");

	if (arg == NULL) {
		DLOG_ERR(LOG_FILT, "filter_stream_init parameters error.");
		return NULL;
	}

	filter_node_t *filter = (filter_node_t *)arg;
	char *thread_arg;

	thread_arg = strdup("[filter: test_1]: thread_arg");

	DLOG_INFO(LOG_FILT, "stream_init: load data: %s | thread data: %s",
			(char *)filter->f_data, thread_arg);

	return thread_arg;
}

/**
 * do filter operations,
 * like a network request, I/O request,
 * and cache request, database request etc.
 */
void *filter_stream_worker(void *thread_arg, void *task_arg)
{
	dlib_buffer_t *buffer;
	DLOG_DEBUG(LOG_FILT, "FILTER(test_1): filter_stream_worker.");

	if (task_arg == NULL || thread_arg == NULL) {
		DLOG_ERR(LOG_FILT, "filter_stream_worker parameters error.");
		return NULL;
	}

	char *worker_data = "[filter: test_1]: worker_data";
	pipeline_data_t *pdata = (pipeline_data_t *)task_arg;
	dserver_conn_t *conn = (dserver_conn_t *)(pdata->p_data);

	if (conn->c_data != NULL) {
		buffer = (dlib_buffer_t *)(conn->c_data);
	} else {
		DLOG_DEBUG(LOG_FILT, "FILTER(test_1): filter_stream_worker init buffer.");
		buffer = dlib_buffer_init(16);
		conn->c_data = (void *)buffer;
	}

	dlib_buffer_t *temp = dlib_buffer_append(buffer, worker_data, strlen(worker_data));
	if (temp != NULL) {
		buffer = temp;
		conn->c_data = buffer;
		pdata->p_data = conn;
	}

	return (void *)pdata;
}

/**
 * (ASYNC with filter_stream_worker)
 * when finish a stream-worker operation,
 * you can deliver data to next filter, 
 * or you can submit data to dispatcher
 */
void *filter_stream_finish(void *res_arg, void *detask_arg)
{
	filter_node_t *filter;
	DLOG_DEBUG(LOG_FILT, "FILTER(test_1): filter_stream_finish.");

	if (res_arg == NULL || detask_arg == NULL) {
		DLOG_ERR(LOG_FILT, "filter_stream_finish parameters error.");
		return NULL;
	}
	pipeline_data_t *pdata = (pipeline_data_t *)res_arg;

	/* deliver to the next filter */
	DLOG_DEBUG(LOG_FILT, "FILTER(test_1): filter_stream_finish deliver to next(test_2).");

	filter = (filter_node_t *)detask_arg;

	pdata->p_filter_flag = FILT_PIPELINE_SUCCESS;
	if (pipeline_deliver_next(filter, pdata) < 0) {
		DLOG_ERR(LOG_FILT, "FILTER(test_1): filter_stream_finish deliver_next error.");
		goto err;	
	}
	
	return NULL;
err:
	pdata->p_filter_flag = FILT_PIPELINE_ERROR;
	pipeline_submit(filter, pdata);
	return NULL;
}

/**
 * (ASYNC with filter_stream_worker)
 * when a stream-worker operation timeouted,
 * you can deliver data to special filter, 
 * or you can submit data to dispatcher and report an error
 */
void *filter_stream_timeout(void *res_arg, void *totask_arg)
{
	filter_node_t *filter;
	if (res_arg == NULL || totask_arg == NULL) {
		DLOG_ERR(LOG_FILT, "filter_stream_timeout parameters error.");
		return NULL;
	}
	pipeline_data_t *pdata = (pipeline_data_t *)res_arg;

	/* deliver to the last filter */
	DLOG_DEBUG(LOG_FILT, "FILTER(test_1): filter_stream_timeout deliver to test_3.");

	filter = filter_get_by_name("test_3");
	if (filter == NULL) {
		DLOG_ERR(LOG_FILT, "FILTER(test_1): filter_stream_timeout get_name error.");
		goto err;	
	}

	pdata->p_filter_flag = FILT_PIPELINE_TIMEOUT;
	if (pipeline_deliver(filter, pdata) < 0) {
		DLOG_ERR(LOG_FILT, "FILTER(test_1): filter_stream_timeout deliver error.");
		goto err;	
	}

	return NULL;
err:
	pdata->p_filter_flag = FILT_PIPELINE_ERROR;
	pipeline_submit(filter, pdata);
	return NULL;
}

/**
 * free persistence resources,
 * like close a persist-connection
 */
void filter_stream_exit(void *thread_arg)
{
	DLOG_DEBUG(LOG_FILT, "FILTER(test_1): filter_stream_exit.");
	if (thread_arg != NULL) {
		dfree(thread_arg);
		thread_arg = NULL;
	}
	return;
}

void filter_on_unload(filter_node_t *filter)
{
	DLOG_DEBUG(LOG_FILT, "FILTER(TEST_1): filter_on_unload.");

	if (filter == NULL) {
		return;
	}
	if (filter->f_data != NULL) {
		dfree(filter->f_data);
		filter->f_data = NULL;
	}
	return;
}

